package com.ehotting.mqbd.core.mq.listener;

import com.ehotting.mqbd.api.exception.SuspendCurrentQueueMomentException;
import com.ehotting.mqbd.core.mq.consumer.action.ConsumerActionAble;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.List;

public class MQListenerOrderly implements MessageListenerOrderly {

    private static final Logger logger = LoggerFactory.getLogger(MQListenerOrderly.class);

    private ConsumerActionAble consumerAction;

    public MQListenerOrderly(ConsumerActionAble consumerAction){
        this.consumerAction = consumerAction;
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
        try{
            for(MessageExt msg : msgs){
                consumerAction.invoke(msg);
            }
        }catch (Exception e){
            if( e instanceof SuspendCurrentQueueMomentException){
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
            logger.error("consumeMessage error.",e);
        }finally {
            MDC.clear();
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
